import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.bean.DwdUserUserLoginBean;
import com.group2.edu.realtime.common.constant.Constant;
import com.group2.edu.realtime.common.util.DateFormatUtil;
import com.group2.edu.realtime.common.util.FlinkSinkUtil;
import javafx.scene.input.DataFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author 申赏赏
 * @date 2024/12/16 10:52<p></p>
 *
 * 需要启动的：zk、kafka、f1、base-log、本程序
 *
 * 可以通过启动kafka消费者:dwd_user_user_login 来查看数据
 */
public class DwdUserUserLogin extends BaseApp {
    public static void main(String[] args) {
        new DwdUserUserLogin().start(10015,
                4,
                "dwd_traffic_page_log",
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> mainSource) {
        // 过滤掉uid为空的数据，并将数据转换成为好处理的JSONObject格式
        SingleOutputStreamOperator<JSONObject> flatDS = mainSource.flatMap(
                new FlatMapFunction<String, JSONObject>() {
                    @Override
                    public void flatMap(String jsonStr, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            if (jsonObj.getJSONObject("common").getString("uid") != null) {
                                out.collect(jsonObj);
                            }
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }

                    }
                }
        );

        // 设置水位线
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = flatDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts");
                                    }
                                }
                        )
        );

        // 按照会话ID分组
        KeyedStream<JSONObject, String> keyedDS = withWatermarkDS
                .keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));

        SingleOutputStreamOperator<JSONObject> processDS = keyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> firstLoginDtState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> firstLoginDtState1 = new ValueStateDescriptor<>("firstLoginDtState", JSONObject.class);
                        firstLoginDtState1.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10L)).build());
                        firstLoginDtState = getRuntimeContext().getState(firstLoginDtState1);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject ValueStateDD = firstLoginDtState.value();
                        Long ts = jsonObj.getLong("ts");
                        if (ValueStateDD == null) {
                            // 当第一条数据到的时候开启定时器
                            firstLoginDtState.update(jsonObj);
                            ctx.timerService().registerEventTimeTimer(ts + 10 * 1000L);
                        } else {
                            Long lastTs = ValueStateDD.getLong("ts");
                            if (ts < lastTs) {
                                firstLoginDtState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        out.collect(firstLoginDtState.value());
                    }
                }
        );

        SingleOutputStreamOperator<String> mapDS = processDS.map(
                new MapFunction<JSONObject, String>() {
                    @Override
                    public String map(JSONObject jsonObj) throws Exception {
                        JSONObject common = jsonObj.getJSONObject("common");
                        Long ts = jsonObj.getLong("ts");
                        String loginTime = DateFormatUtil.tsToDateTime(ts);
                        String dateId = loginTime.substring(0, 10);
                        DwdUserUserLoginBean dwdUserUserLoginBean = DwdUserUserLoginBean.builder()
                                .userId(common.getString("uid"))
                                .dateId(dateId)
                                .loginTime(loginTime)
                                .sourceId(common.getString("sc"))
                                .channel(common.getString("ch"))
                                .provinceId(common.getString("ar"))
                                .versionCode(common.getString("vc"))
                                .midId(common.getString("mid"))
                                .brand(common.getString("ba"))
                                .model(common.getString("md"))
                                .operatingSystem(common.getString("os"))
                                .ts(ts)
                                .build();
                        return JSON.toJSONString(dwdUserUserLoginBean);
                    }
                }
        );

        // 数据写入kafka的主题中
        String sinkTopic = "dwd_user_user_login";
        mapDS.sinkTo(FlinkSinkUtil.getKafkaSink(sinkTopic));



        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}


//{
// "common":{"ar":"1","ba":"iPhone","ch":"Appstore","is_new":"0","md":"iPhone X","mid":"mid_239",
// "os":"iOS 13.2.3","sc":"2","sid":"79dd651d-9d2a-4f5a-8253-6edfef9debbf","uid":"2043","vc":"v2.1.134"},
// "displays":[{"display_type":"promotion","item":"4","item_type":"course_id","order":1,"pos_id":2},
// {"display_type":"promotion","item":"10","item_type":"course_id","order":2,"pos_id":4},
// {"display_type":"query","item":"4","item_type":"course_id","order":3,"pos_id":3},
// {"display_type":"query","item":"9","item_type":"course_id","order":4,"pos_id":5}],
// "page":{"during_time":12274,"item":"54468","item_type":"order_id","last_page_id":"order","page_id":"payment"},
// "ts":1734314213971
// }
